/**
 * 
 */
package name.yzhu.wbdata.server;

import java.io.IOException;

import javax.servlet.ServletException;
import javax.servlet.UnavailableException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import name.yzhu.cloud.CloudException;
import name.yzhu.cloud.hbase.HBaseResult;
import name.yzhu.cloud.hbase.HBaseScan;
import name.yzhu.cloud.hbase.NoFoundHBaseRecord;
import name.yzhu.cloud.hbase.SingleTableReader;
import name.yzhu.cloud.hbase.YamlHBaseRowBase;
import name.yzhu.httpserver.AbstractHttpServlet;
import name.yzhu.httpserver.HttpServerUtil;
import name.yzhu.wbdata.Constant;
import name.yzhu.wbdata.io.WeiboYamlRow;

import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;

/**
 * @author yhzhu
 *
 */
public class HBaseServelt extends AbstractHttpServlet {
	private static Logger LOG = Logger.getLogger(HBaseServelt.class);
	/**
	 * 
	 */
	private static final long serialVersionUID = 6217916979367365662L;
	private static final int MAX_RETRIEVE_RECORD = 30;
	private SingleTableReader reader = null;
	private Configuration hbaseCFG = null;

	public HBaseServelt(Configuration hbaseCFG) {
		this.hbaseCFG = hbaseCFG;
	}
	
	/* (non-Javadoc)
	 * @see name.yzhu.httpserver.AbstractHttpServlet#getContextRoot()
	 */
	@Override
	public String getContextRoot() {
		return "/";
	}

	/* (non-Javadoc)
	 * @see name.yzhu.httpserver.AbstractHttpServlet#getContextPath()
	 */
	@Override
	public String getContextPath() {
		return "/hbase";
	}

	/* (non-Javadoc)
	 * @see name.yzhu.httpserver.AbstractHttpServlet#process(javax.servlet.http.HttpServletRequest, javax.servlet.http.HttpServletResponse)
	 */
	@Override
	protected void process(HttpServletRequest req, HttpServletResponse resp)
			throws ServletException, IOException {
		if (null != req.getParameter("key"))
			getRow(req, resp);
		else if (null != req.getParameter("data_source") &&
				 null != req.getParameter("timeline"))
			getResultWithDataSouce(req, resp);
		else
			getAllResult(req,resp);
	}
	
	private void getAllResult(HttpServletRequest req,
			HttpServletResponse resp) throws ServletException, IOException {
		int max = MAX_RETRIEVE_RECORD;
		if (null != req.getParameter("max"))
		   max = Integer.valueOf(req.getParameter("max")).intValue();
		
		HBaseScan scan = new HBaseScan();
		HBaseResult rs = null;
		try {
			rs = reader.getHBaseResult(scan);
			YamlHBaseRowBase row = new WeiboYamlRow();
			int count = 0;
			while (rs.next(row)) {
				LOG.debug(row.Dump());
				resp.getWriter().print(HttpServerUtil.format("hbase search", HttpServerUtil.yamlToHtml(row)));
				if (++count >= max)
					break;
			}
		} catch (CloudException e) {
			LOG.error(e.getMessage(), e);
			throw new ServletException(e);
		} catch (IOException e) {
			LOG.error(e.getMessage(), e);
			throw new IOException(e);
		}
		finally {
			rs.close();
		}
	}

	private void getResultWithDataSouce(HttpServletRequest req,
			HttpServletResponse resp) throws ServletException, IOException {
		String dataSource = req.getParameter("data_source");
		Integer timeline = Integer.valueOf(req.getParameter("timeline"));
		int max = MAX_RETRIEVE_RECORD;
		if (null != req.getParameter("max"))
		   max = Integer.valueOf(req.getParameter("max")).intValue();
		
		HBaseScan scan = new HBaseScan();
		scan.addFilter(name.yzhu.cloud.Constant.DEFAULT_FAMILY_NAME,
				name.yzhu.common.Constant.YAML_FIELD_DATA_SOURCE,
				dataSource);
		scan.addFilter(name.yzhu.cloud.Constant.DEFAULT_FAMILY_NAME,
				Constant.YAML_FIELD_WEIBO_TIME_LINE,
				timeline);
		HBaseResult rs  = null;
		try {
			rs = reader.getHBaseResult(scan);
			YamlHBaseRowBase row = new WeiboYamlRow();
			int count = 0;
			while (rs.next(row)) {
				LOG.debug(row.Dump());
				resp.getWriter().print(HttpServerUtil.format("hbase search", HttpServerUtil.yamlToHtml(row)));
				if (++count >= max)
					break;
			}
		} catch (CloudException e) {
			LOG.error(e.getMessage(), e);
			throw new ServletException(e);
		} catch (IOException e) {
			LOG.error(e.getMessage(), e);
			throw new IOException(e);
		}
		finally {
			rs.close();
		}
	}

	private void getRow(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException{
		String key = req.getParameter("key");
		if (key == null){
			resp.getWriter().println("please specify key");
			return ;
		}
		YamlHBaseRowBase row = new WeiboYamlRow(key);
		
		try {
			reader.read(row );
			LOG.debug(row.Dump());
			resp.getWriter().print(HttpServerUtil.format("hbase search", HttpServerUtil.yamlToHtml(row)));
		}catch (NoFoundHBaseRecord e) {
			LOG.debug(" No find: "+key, e);
			resp.getWriter().println("No find this rowKey: "+key);
			throw new ServletException(e);
		} 
		catch (CloudException e) {
			LOG.error(e.getMessage()+" rowKey: "+key, e);
			throw new ServletException(e);
		}
	}

	@Override
	public void init() throws UnavailableException {
		super.init();
		reader  = new SingleTableReader(hbaseCFG , Constant.HBASE_WEIBO_TABLE);
	}

	@Override
	public void destroy() {
		try {
			reader.close();
		} catch (CloudException e) {
			LOG.error(e.getMessage(), e);
		}
		super.destroy();
	}

	
}
